use crate::physical_plan::window::accumulator::{PhysicalPlanExprAccumulator, PhysicalPlanExprAccumulatorComb, accumulator_by_name};
use crate::physical_plan::PhysicalPlanExprArg;
use crate::run_plan::func::RunPlanExprFunc;
use crate::data_frame::single_data::SingleData;
use crate::run_plan::{HandResult};
use sapeur::logical_comb::LogicalCombOutput;
use crate::logical_plan::expr::func::{LogicalPlanExprFuncOrValue, logical_plan_expr_func_map_to_run_plan_expr_func};
use crate::conf_init::{MapResult};
use crate::data_frame::DataInstance;
use crate::logical_plan::expr::accumulator::{LogicalPlanExprAccumulator};

#[derive(Debug,Clone)]
pub enum RunPlanExprAccumulatorArg{
    Normal(Box<PhysicalPlanExprArg>),
    OtherRunPlan(Box<RunPlanExprFunc>),
}

#[derive(Debug)]
pub struct RunPlanExprAccumulator<D,S>{
    physical_accumulator: Box<dyn PhysicalPlanExprAccumulator<D,S>>,
    args: Vec<RunPlanExprAccumulatorArg>,
}

impl <D,S> Clone for RunPlanExprAccumulator<D,S>{
    fn clone(&self) -> Self{
        let mut args = Vec::new();
        for t in self.args.iter(){
            args.push(t.clone());
        }
        let pa = self.physical_accumulator._clone_self();
        return RunPlanExprAccumulator{
            physical_accumulator: pa,
            args: args,
        };
    }
}

impl <D,S> RunPlanExprAccumulator<D,S>{
    fn new(physical_paln: Box<dyn PhysicalPlanExprAccumulator<D,S>>,
           args: Vec<RunPlanExprAccumulatorArg>)->Box<Self>{

        Box::new(
            RunPlanExprAccumulator{
                physical_accumulator: physical_paln,
                args: args,
            }
        )
    }
    async fn args(&mut self, data: &mut DataInstance)->Result<Vec<Option<Box<PhysicalPlanExprArg>>>,()>{
        let mut args = Vec::new();

        for a in self.args.iter_mut(){
            match a{
                RunPlanExprAccumulatorArg::Normal(a) => {args.push(Some(a.clone()));},
                RunPlanExprAccumulatorArg::OtherRunPlan(ref mut f) => {
                    let ret = f.evaluate(data).await;
                    for r in ret.into_iter(){
                        if let HandResult::Ok(r) = r{
                            args.push(Some(PhysicalPlanExprArg::from_real_value(r)));
                        }else{
                            args.push(None);
                        }
                    }
                },
            }
        }
        return Ok(args);
    }
    async fn args_single(&mut self, data: &mut SingleData)->Result<Vec<Box<PhysicalPlanExprArg>>,()>{
        let mut args = Vec::new();

        for a in self.args.iter_mut(){
            match a{
                RunPlanExprAccumulatorArg::Normal(a) => {args.push(a.clone());},
                RunPlanExprAccumulatorArg::OtherRunPlan(ref mut f) => {
                    let ret = f.evaluate_single(data).await;
                    if let HandResult::Ok(r) = ret {
                        args.push(PhysicalPlanExprArg::from_real_value(r));
                    }else{
                        return Err(());
                    }
                },
            }
        }
        return Ok(args);
    }
    async fn args_no_func(&mut self)->Result<Vec<Box<PhysicalPlanExprArg>>,()>{
        let mut args = Vec::new();

        for a in self.args.iter_mut(){
            match a{
                RunPlanExprAccumulatorArg::Normal(a) => {
                    args.push(a.clone());
                },
                RunPlanExprAccumulatorArg::OtherRunPlan(_) => {
                    return Err(());
                },
            }
        }

        return Ok(args);
    }
}

impl RunPlanExprAccumulator<SingleData,Box<PhysicalPlanExprArg>>{
    async fn _add(&mut self, mut data: SingleData) ->Result<(),()>{
        let args = self.args_single(&mut data).await?;

        return self.physical_accumulator._add(args.as_slice(), data).await;
    }
    async fn _sub(&mut self, mut data: SingleData) ->Result<(),()>{
        let args = self.args_single(&mut data).await?;

        return self.physical_accumulator._sub(args.as_slice(), data).await;
    }
    async fn _check(&mut self)->Result<bool,()>{
        return self.physical_accumulator._check().await;
    }
    fn _clone_self(&self) ->Self{
        let mut args: Vec<RunPlanExprAccumulatorArg> = Vec::new();

        for t in self.args.iter(){
            args.push(t.clone());
        }

        return RunPlanExprAccumulator{
            physical_accumulator: self.physical_accumulator._clone_self(),
            args: args,
        };
    }
}

#[derive(Clone)]
pub enum RunPlanExprAccumulatorComb<D,S>{
    And(Box<RunPlanExprAccumulatorComb<D,S>>, Box<RunPlanExprAccumulatorComb<D,S>>),
    Or(Box<RunPlanExprAccumulatorComb<D,S>>, Box<RunPlanExprAccumulatorComb<D,S>>),
    Single(Box<RunPlanExprAccumulator<D,S>>),
}

#[async_trait]
impl PhysicalPlanExprAccumulatorComb<SingleData> for RunPlanExprAccumulatorComb<SingleData,Box<PhysicalPlanExprArg>>{
    async fn add(&mut self, data: SingleData)->Result<(),()>{
        match self{
            RunPlanExprAccumulatorComb::And(left,right) => {
                left.add(data.clone()).await?;
                right.add(data).await
            },
            RunPlanExprAccumulatorComb::Or(left, right) => {
                left.add(data.clone()).await?;
                right.add(data).await
            },
            RunPlanExprAccumulatorComb::Single(single) => {
                single._add(data).await
            },
        }
    }
    async fn sub(&mut self, mut data: SingleData)->Result<(),()>{
        match self{
            RunPlanExprAccumulatorComb::And(left,right) => {
                left.sub(data.clone()).await?;
                right.sub(data).await
            },
            RunPlanExprAccumulatorComb::Or(left, right) => {
                left.sub(data.clone()).await?;
                right.sub(data).await
            },
            RunPlanExprAccumulatorComb::Single(single) => {
                let args = single.args_single(&mut data).await?;
                single._sub(data).await
            },
        }
    }
    async fn check(&mut self)->Result<bool,()>{
        match self{
            RunPlanExprAccumulatorComb::And(left,right) => {
                Ok(left.check().await? & right.check().await?)
            },
            RunPlanExprAccumulatorComb::Or(left, right) => {
                Ok(left.check().await? | right.check().await?)
            },
            RunPlanExprAccumulatorComb::Single(single) => {
                single._check().await
            },
        }
    }
    fn clone_self(&self) ->Self{
        match self{
            RunPlanExprAccumulatorComb::And(left,right) => {
                RunPlanExprAccumulatorComb::And(Box::new(left.clone_self()), Box::new(right.clone_self()))
            },
            RunPlanExprAccumulatorComb::Or(left, right) => {
                RunPlanExprAccumulatorComb::Or(Box::new(left.clone_self()), Box::new(right.clone_self()))
            },
            RunPlanExprAccumulatorComb::Single(single) => {
                RunPlanExprAccumulatorComb::Single(single.clone())
            },
        }
    }
}

pub fn logical_plan_accumulator_comb_to_run_plan(accumulator: &LogicalCombOutput<LogicalPlanExprAccumulator>)
    ->MapResult<Box<RunPlanExprAccumulatorComb<SingleData,Box<PhysicalPlanExprArg>>>>{

    match accumulator{
        LogicalCombOutput::Single(single) => {

            let single = logical_plan_accumulator_to_run_plan(single)?;
            return MapResult::Ok(Box::new(RunPlanExprAccumulatorComb::Single(single)));
        },
        LogicalCombOutput::And(left,right) => {
            let left = logical_plan_accumulator_comb_to_run_plan(left)?;
            let right = logical_plan_accumulator_comb_to_run_plan(right)?;
            return MapResult::Ok(Box::new(RunPlanExprAccumulatorComb::And(left,right)));
        },
        LogicalCombOutput::Or(left, right) => {
            let left = logical_plan_accumulator_comb_to_run_plan(left)?;
            let right = logical_plan_accumulator_comb_to_run_plan(right)?;
            return MapResult::Ok(Box::new(RunPlanExprAccumulatorComb::Or(left,right)));
        },
    }
}

pub fn logical_plan_accumulator_to_run_plan(conf: &LogicalPlanExprAccumulator)
                        ->MapResult<Box<RunPlanExprAccumulator<SingleData,Box<PhysicalPlanExprArg>>>>{


    let mut value_list = Vec::new();
    for arg in conf.2.iter(){
        match &arg{
            LogicalPlanExprFuncOrValue::Value(arg) => {
                let physical_expr = PhysicalPlanExprArg::from(arg.clone());
                value_list.push(RunPlanExprAccumulatorArg::Normal(physical_expr));
            },
            LogicalPlanExprFuncOrValue::FuncWithValue(_func_name, _func_args) => {
                let other_func = logical_plan_expr_func_map_to_run_plan_expr_func(&arg)?;
                value_list.push(RunPlanExprAccumulatorArg::OtherRunPlan(other_func));
            },
        }
    }

    let physical_accumulator = accumulator_by_name(conf.0.as_str(),
                                                   conf.1.as_slice())?;

    return MapResult::Ok(RunPlanExprAccumulator::new(physical_accumulator, value_list));
}


